/*
 * Copyright (c) 2020 - present, Inspur Genersoft Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.iec.edp.caf.rpc;

import io.iec.edp.caf.commons.utils.SpringBeanUtils;
import io.iec.edp.caf.rpc.api.InternalSvrContainer;
import io.iec.edp.caf.rpc.api.entity.RpcServiceDefinition;
import io.iec.edp.caf.rpc.api.local.InternalServiceLocalStorage;
import io.iec.edp.caf.rpc.api.utils.DefinitionUtil;
import io.iec.edp.caf.rpc.client.RpcClassHolder;
import io.iec.edp.caf.rpc.registry.api.RpcDefinitionRegistry;

import java.util.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * RPC服务注册器
 *
 * @author guowenchang
 * @date 2020-11-16
 */
public class RpcServiceRegister {
    private static final RpcServiceRegister singleton = new RpcServiceRegister();

    private Boolean rpcRegistryFinished = false;

    //rpc服务注册表
    private final Map<Class<?>, Object> rpcServiceMap = new HashMap<>();

    //rpc服务注册
    private RpcDefinitionRegistry rpcDefinitionRegistry;

    private Boolean cloudRegistry = false;

    private final Lock lock = new ReentrantLock();

    /**
     * 获取单例对象
     *
     * @return
     */
    public static RpcServiceRegister getSingleton() {
        return singleton;
    }

    public void setCloudRegistry() {
        cloudRegistry = true;
    }

    /**
     * 注册Rpc服务
     *
     * @param rpcInterface
     * @param rpcService
     * @param <T>
     */
    public <T> void registerRpcService(Class<T> rpcInterface, T rpcService) {
        try {
            lock.lock();

            doRegisterRpcService(rpcInterface, rpcService);
            //这个判断是因为在RpcStartupCompletedListener里，平台内部统一完成了RPC的注册后，会把rpcRegistryFinished设置为true
            //这样在统一注册后在继续通过该接口注册服务时，就实时的发布出来RPC接口，如果还没有统一注册，则由RpcStartupCompletedListener统一注册一次即可
            if (rpcRegistryFinished) {
                //直接发布RpcInterface
                doPublishRpcService(rpcInterface, rpcService);
            }
        } finally {
            lock.unlock();
        }
    }

    private <T> void doPublishRpcService(Class<T> rpcInterface, T rpcService) {
        //根据Interface获取Rpc服务定义
        RpcServiceDefinition rpcServiceDefinition = DefinitionUtil.transformServiceDefinition(rpcInterface);

        //将服务定义添加到列表
        InternalServiceLocalStorage.current.getRpcServiceDefinitions().add(rpcServiceDefinition);
        //注册到Rpc服务中心
        if (rpcDefinitionRegistry == null) {
            rpcDefinitionRegistry = SpringBeanUtils.getBean(RpcClassHolder.class).getRpcClass(RpcDefinitionRegistry.class);
        }

        rpcDefinitionRegistry.register(Collections.singletonList(rpcServiceDefinition));

        //这句话有用，在本地调用的时候如果bean获取不到（并行启动模式下），会从这个缓存里拿服务
        InternalSvrContainer.putService(rpcInterface, rpcService);
    }

    private <T> void doRegisterRpcService(Class<T> rpcInterface, T rpcService) {
        //如果重复发布了同一个RPC接口的实现 这里报错
        Object duplicateService = rpcServiceMap.get(rpcInterface);
        if (duplicateService == null) {
            rpcServiceMap.put(rpcInterface, rpcService);
            //这句话有用，在本地调用的时候如果bean获取不到（并行启动模式下），会从这个缓存里拿服务
            InternalSvrContainer.putService(rpcInterface,rpcService);
        } else {
            throw new RuntimeException("Duplicated RPC service found: " + rpcService.getClass().getName()
                    + " " + duplicateService.getClass().getName());
        }
    }

    public void finishRegistry() {
        this.rpcRegistryFinished = true;
    }

    public Set<Class<?>> getRpcInterfaces() {
        return rpcServiceMap.keySet();
    }
}
